package com.poetic.market.analysis;

import com.poetic.market.analysis.aggregate.CountAgg;
import com.poetic.market.analysis.domain.AdClickLog;
import com.poetic.market.analysis.domain.CountByProvince;
import com.poetic.market.analysis.function.CountResult;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * <pre>
 *  页面广告点击量统计
 *  根据province 进行 keyBy，然后开一小时的时间窗口，
 * 滑动距离为 5 秒，统计窗口内的点击事件数量
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class AdStatisticsByGeoTask {
    public static void main(String[] args) throws Exception {
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/AdClickLog.csv";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        DataStreamSource<String> dataStream = env.readTextFile(path);

        dataStream.map(new MapFunction<String, AdClickLog>() {
            @Override
            public AdClickLog map(String value) throws Exception {
                String[] data = value.split("\\,");
                AdClickLog adClickLog = AdClickLog.builder()
                        .userId(Long.valueOf(data[0]))
                        .adId(Long.valueOf(data[1]))
                        .province(data[2])
                        .city(data[3])
                        .timestamp(Long.valueOf(data[4]))
                        .build();
                return adClickLog;
            }
        })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<AdClickLog>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner(new SerializableTimestampAssigner<AdClickLog>() {
                    @Override
                    public long extractTimestamp(AdClickLog element, long recordTimestamp) {
                        return element.getTimestamp();
                    }
                }))
                .keyBy(new KeySelector<AdClickLog, String>() {
                    @Override
                    public String getKey(AdClickLog value) throws Exception {
                        return value.getProvince();
                    }
                })
                .timeWindow(Time.hours(1), Time.seconds(5))
                /**
                 * CountByProvince(windowEnd=1970/01/18-19:54:20, province=beijing, count=2)
                 * CountByProvince(windowEnd=1970/01/18-19:54:20, province=guangdong, count=5)
                 */
                .aggregate(new CountAgg(), new CountResult())
                /**
                 * 方法二
                 * CountByProvince(windowEnd=1970/01/18-19:54:20, province=beijing, count=2)
                 * CountByProvince(windowEnd=1970/01/18-19:54:20, province=guangdong, count=5)
                 */
//                .process(new ProcessWindowFunction<AdClickLog, CountByProvince, String, TimeWindow>() {
//                    @Override
//                    public void process(String key, Context context, Iterable<AdClickLog> elements, Collector<CountByProvince> out) throws Exception {
//                        SimpleDateFormat format = new SimpleDateFormat("yyyy/MM/dd-HH:mm:ss");
//                        CountByProvince countByProvince = CountByProvince.builder()
//                                .windowEnd(format.format(new Date(context.window().getEnd())))
//                                .province(key)
//                                .count(elements.spliterator().estimateSize())
//                                .build();
//
//                        out.collect(countByProvince);
//                    }
//                })
                .print();

        env.execute("ad statistics by geo task");

    }
}
